import torch
import torch.nn as nn

# Given targets and network output
targets = torch.tensor([1, 4, 5, 8], dtype=torch.float32)
network_output = torch.tensor([0.1, 4.4, 0.2, 10], dtype=torch.float32)

# Compute HuberLoss
huber_loss = nn.HuberLoss(reduction='mean', delta=1.0)
loss_value = huber_loss(network_output, targets)
loss_value.item()

1.571250081062317

import os
import pandas as pd
import torch
from torchvision.io import read_image
from torchvision import transforms
from torch.utils.data import Dataset

class InsectsDataset(Dataset):
    def __init__(self, csv_file, root_dir, transform=None):
        """
        Args:
            csv_file (string): Path to the csv file with annotations.
            root_dir (string): Directory with all the images.
            transform (callable, optional): Optional transform to be applied on a sample.
        """
        self.insects_frame = pd.read_csv(csv_file)
        self.root_dir = root_dir
        self.transform = transform

    def __len__(self):
        return len(self.insects_frame)

    def __getitem__(self, idx):
        img_name = os.path.join(self.root_dir, self.insects_frame.iloc[idx, 2])  # Assuming filename is in the second col
        image = read_image(img_name)  # Using torchvision.io to read the image
        species = self.insects_frame.iloc[idx, 1]  # Assuming species is in the first column

        # Transform picture to correct size
        if self.transform:
            image = self.transform(image)

        return [image, species]


# Make a conditional transform to tensor (since we ue torchvision.io to read the image, it will already be a tensor)
class ConditionalToTensor(transforms.ToTensor):
    def __call__(self, pic):
        if isinstance(pic, torch.Tensor):
            return pic
        return super().__call__(pic)

# Define a basic transform to convert the images to PyTorch tensors
transform = transforms.Compose([
    transforms.Resize((128, 128)), # Resize images for demonstration purposes
    ConditionalToTensor() # Checks if the input is a tensor before converting it
])

csv_file='./data/Insects.csv'
Image_folder='./data/Insects/'

# Create an instance of the dataset
insects_dataset = InsectsDataset(csv_file=csv_file, root_dir=Image_folder, transform=transform)

insects_dataset.__getitem__(0)

[tensor([[[112, 109, 105,  ..., 117, 108, 109],
          [111, 108, 107,  ..., 119, 109, 109],
          [111, 108, 109,  ..., 120, 107, 109],
          ...,
          [115, 106, 106,  ..., 109, 108, 103],
          [116, 107, 106,  ..., 112, 108, 105],
          [117, 110, 107,  ..., 110, 106, 105]],
 
         [[137, 138, 136,  ..., 147, 135, 136],
          [140, 139, 136,  ..., 146, 137, 136],
          [139, 136, 137,  ..., 143, 138, 139],
          ...,
          [145, 136, 136,  ..., 140, 137, 131],
          [146, 137, 135,  ..., 142, 137, 133],
          [144, 139, 137,  ..., 141, 137, 134]],
 
         [[201, 201, 200,  ..., 202, 202, 205],
          [201, 199, 200,  ..., 209, 202, 202],
          [202, 200, 200,  ..., 206, 202, 201],
          ...,
          [193, 196, 196,  ..., 195, 196, 194],
          [194, 197, 195,  ..., 200, 197, 198],
          [199, 199, 201,  ..., 198, 193, 199]]], dtype=torch.uint8),
 'Andrena fulva']

def custom_dataLoader():
    # Detect if a GPU is available
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

    # Define a basic transform to convert the images to PyTorch tensors
    transform = transforms.Compose([
        transforms.Resize((128, 128)), # Resize images for demonstration purposes
        ConditionalToTensor() # Checks if the input is a tensor before converting it
    ])

    csv_file='./data/Insects.csv'
    Image_folder='./data/Insects/'

    # Create an instance of the dataset
    insects_dataset = InsectsDataset(csv_file=csv_file, root_dir=Image_folder, transform=transform)

    # Set up the dataset.
    trainloader = torch.utils.data.DataLoader(insects_dataset, 
                                            batch_size=4, 
                                            shuffle=True, 
                                            num_workers=2)

    # get some images
    dataiter = iter(trainloader)
    images, labels = dataiter.next()

    # Move data (tensors) to the GPU
    images = images.to(device)

    # Calculate total number of images to be displayed
    total_images = 5 * 4  # 5 batches of size 4

    # Adjusted for 3 columns (we only want 1 extra row if there are unused subplot axes at the end)
    nrows = total_images // 3 + (total_images % 3 > 0) 
    
    fig, axs = plt.subplots(nrows, 3, figsize=(9, 3 * nrows))  # Adjusted figsize for 3 columns

    img_count = 0
    for i in range(5): # Run through 5 batches
        images, labels = dataiter.next()
        images = images.to(device) # Move the tensors to the GPU
        for image, label in zip(images, labels): # Run through all samples in a batch
            row = img_count // 3
            col = img_count % 3
            axs[row, col].imshow(np.transpose(image.cpu().numpy(), (1, 2, 0)))
            axs[row, col].set_title(label)
            axs[row, col].axis('off')
            img_count += 1

    # If there are unused subplot axes at the end, turn them off
    for idx in range(img_count, nrows * 3):
        row = idx // 3
        col = idx % 3
        axs[row, col].axis('off')

    plt.tight_layout()
    plt.show()

# For multiprocessing to work, the code needs to be in a separate .py file and we need to run it with 
# the __name__ == '__main__' check for Windows

# Load the multiprocessing library DataLoader, and make it reloadable for devolopment purposes
import functions.Custom_DataLoader
from importlib import reload
reload(functions.Custom_DataLoader)
from functions.Custom_DataLoader import custom_dataLoader

if __name__ == '__main__':
    # Call the dataloader function
    custom_dataLoader()

# import pandas as pd
import matplotlib.pyplot as plt

# Load the datasets
train_data = pd.read_csv('./data/trainData.txt', header=None, delim_whitespace=True, 
                         names=['label', 'feature1', 'feature2'])
test_data = pd.read_csv('./data/testData.txt', header=None, delim_whitespace=True, 
                        names=['label', 'feature1', 'feature2'])

# Define a normalization based on label values (for plotting purposes)
norm = plt.Normalize(train_data['label'].min(), train_data['label'].max())

# Visualize the data
plt.figure(figsize=(8, 6))
for label in sorted(train_data['label'].unique()):
    subset = train_data[train_data['label'] == label]
    plt.scatter(subset['feature1'], subset['feature2'], norm=norm, c=subset['label'], 
                cmap="jet", label=f'Label {int(label)}', alpha=0.6, edgecolors='w')
    
plt.title('Visualization of the Training Data')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')
plt.xlim(-1.15, 1.15)
plt.ylim(-1.15, 1.15)
plt.legend(loc='upper right')
plt.grid(True)

plt.show()

import torch
import numpy as np
import torch.nn as nn
import torch.optim as optim

# Extracting features and labels from the dataframe
trainX = train_data[["feature1", "feature2"]].values
trainLabels = train_data["label"].values.astype(int)
testX = test_data[["feature1", "feature2"]].values
testLabels = test_data["label"].values.astype(int)

# Convert them to PyTorch tensors
trainX_tensor = torch.tensor(trainX, dtype=torch.float32)
trainLabels_tensor = torch.tensor(trainLabels, dtype=torch.int64)
testX_tensor = torch.tensor(testX, dtype=torch.float32)
testLabels_tensor = torch.tensor(testLabels, dtype=torch.int64)

# Define the neural network
class SimpleNN(nn.Module):
    def __init__(self, input_dim, hidden_dim1, hidden_dim2, output_dim):
        super(SimpleNN, self).__init__()

        # Hidden Layer 1
        self.fc1 = nn.Linear(input_dim, hidden_dim1)
        # Hidden Layer 2
        self.fc2 = nn.Linear(hidden_dim1, hidden_dim2)
        # Output Layer
        self.fc3 = nn.Linear(hidden_dim2, output_dim)
    
    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# Define network dimensions
input_dim = trainX.shape[1]
hidden_dim1 = 64
hidden_dim2 = 32
output_dim = len(np.unique(trainLabels))

# Instantiate the model
model = SimpleNN(input_dim, hidden_dim1, hidden_dim2, output_dim)

model

SimpleNN(
  (fc1): Linear(in_features=2, out_features=64, bias=True)
  (fc2): Linear(in_features=64, out_features=32, bias=True)
  (fc3): Linear(in_features=32, out_features=5, bias=True)
)

def k_fold_split(dataset, k_folds=5):
    
    # Shuffle dataset indices
    indices = torch.randperm(len(dataset))
    
    # Split indices into k chunks/folds
    return [indices[i::k_folds] for i in range(k_folds)]

from torch.utils.data import Subset, DataLoader, TensorDataset

import random
import numpy as np
import torch

# Set seed for reproducibility
def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

set_seed(42)

# Define the loss function
criterion = nn.CrossEntropyLoss()

# Lists to keep track of training and validation losses
train_losses = []
test_losses = []

# Parameters
batch_size = 50
epochs = 500
k_folds = 5
early_stop_epochs = 30  # Number of epochs to wait before stopping if no improvement in validation loss

# Convert the training data into a TensorDataset and use DataLoader for minibatch handling
train_dataset = TensorDataset(trainX_tensor, trainLabels_tensor)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# Lists to keep track of training and validation losses across folds
all_train_losses = []
all_val_losses = []

# Initialize early stopping counter and best validation loss across folds
best_val_loss = float('inf')
best_model_weights = None
fold_val_losses = []

# Get the k-fold split indices
k_fold_indices = k_fold_split(train_dataset, k_folds)

# A list to store the best model weights for each fold
best_weights_per_fold = []

for fold, val_indices in enumerate(k_fold_indices):
    print(f"FOLD {fold + 1}")
    
    # Create training indices by excluding the validation indices
    train_indices = [i for i in range(len(train_dataset)) if i not in val_indices]
    
    # Create training and validation subsets
    train_subset = Subset(train_dataset, train_indices)
    val_subset = Subset(train_dataset, val_indices)
    
    train_loader = DataLoader(train_subset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_subset, batch_size=batch_size, shuffle=False)
    
    # Reset the learning rate for each fold
    lr = 0.01

    # Reset the model and optimizer for each fold
    model = SimpleNN(input_dim, hidden_dim1, hidden_dim2, output_dim)
    optimizer = optim.Adam(model.parameters(), lr=lr)
    
    # Lists to keep track of training and validation losses
    train_losses = []
    val_losses = []

    # Initialize early stopping counter and best validation loss for this fold
    early_stop_counter = 0
    fold_best_val_loss = float('inf')
    
    # Train the model
    for epoch in range(epochs):
        # Training loop for one epoch (one full pass over the training set minus the validation indices)
        for batch_X, batch_labels in train_loader:

            # Compute the model's predictions on the test/validation data.
            # Here, the model processes the train data through our model defined above and returns 
            # its predictions.
            outputs = model(batch_X)

            # Compute the loss between the model's predictions and the actual labels in the 
            # test/validation data.
            # This gives us an idea of how well the model is predicting on trained data.
            loss = criterion(outputs, batch_labels)
            
            # Backward pass and optimization

            # Zero out the gradients for the optimizer. This prevents accumulation of gradients from 
            # previous iterations.
            # In PyTorch, gradients accumulate by default (useful for things like RNNs), so they need 
            # to be cleared out 
            # before each new gradient computation.
            optimizer.zero_grad()

            # Compute the gradients for each parameter based on the current loss. 
            # This function backpropagates the error through the network by computing the gradient of 
            # the loss with 
            # respect to each (learnable) parameter.
            loss.backward()

            # Update the model's parameters based on the computed gradients. Our model was defined 
            # in the previous cell,
            # This step adjusts each parameter (like weights and biases) in the direction that reduces 
            # the loss.
            # Here we namely use the ADAM optimizer.
            optimizer.step()
        
        # Training loss for the epoch
        outputs = model(trainX_tensor[train_indices])
        train_loss = criterion(outputs, trainLabels_tensor[train_indices])
        train_losses.append(train_loss.item())
        
        # Validation loss

        # `torch.no_grad()` is a context manager that disables gradient computation during inference.
        # Disabling gradient computation is beneficial for evaluation, saves memory, and speeds up the 
        # computation.
        # We don't need gradients during validation or testing since we're not updating the model parameters.
        with torch.no_grad():

            # Here, the model processes the test data and returns its predictions.
            outputs = model(trainX_tensor[val_indices])

            # This gives us an idea of how well the model is performing on unseen data.
            val_loss = criterion(outputs, trainLabels_tensor[val_indices])

            # Record the validation loss
            val_losses.append(val_loss.item())

        # Check early stopping (to prevent overfitting and speed up training)
        if val_loss < fold_best_val_loss:
            fold_best_val_loss = val_loss
            early_stop_counter = 0 # Reset early stopping counter

            # Save the best weights for this fold
            current_best_weights = model.state_dict().copy()
        else:
            early_stop_counter += 1 # No improvement, increase early stopping counter
            if early_stop_counter >= early_stop_epochs:
                print(f"Early stopping on epoch {epoch+1}")
                break
        
        # Print loss every 50 epochs
        if (epoch + 1) % 50 == 0:
            # Decay the learning rate every 50 epochs
            lr = lr * 0.8
            print(f'Epoch [{epoch + 1}/{epochs}], Train Loss: {train_loss.item():.4f},',
                  f'Val Loss: {val_loss.item():.4f}, Learning Rate: {lr:.4f}')
    
    # Save the best weights for this fold
    best_weights_per_fold.append(current_best_weights)

    # Save the training and validation losses for this fold
    fold_val_losses.append(fold_best_val_loss)
    all_train_losses.append(train_losses)
    all_val_losses.append(val_losses)

# Choose the best model based on average validation loss across folds
best_fold = np.argmin(fold_val_losses)
best_model_weights = best_weights_per_fold[best_fold]

FOLD 1
Epoch [50/500], Train Loss: 0.0400, Val Loss: 0.0364, Learning Rate: 0.0080
Epoch [100/500], Train Loss: 0.0426, Val Loss: 0.0381, Learning Rate: 0.0064
Epoch [150/500], Train Loss: 0.0163, Val Loss: 0.0114, Learning Rate: 0.0051
Epoch [200/500], Train Loss: 0.0223, Val Loss: 0.0094, Learning Rate: 0.0041
Early stopping on epoch 203
FOLD 2
Epoch [50/500], Train Loss: 0.0466, Val Loss: 0.0661, Learning Rate: 0.0080
Epoch [100/500], Train Loss: 0.0223, Val Loss: 0.0841, Learning Rate: 0.0064
Early stopping on epoch 108
FOLD 3
Epoch [50/500], Train Loss: 0.0949, Val Loss: 0.1087, Learning Rate: 0.0080
Epoch [100/500], Train Loss: 0.0159, Val Loss: 0.0418, Learning Rate: 0.0064
Early stopping on epoch 126
FOLD 4
Epoch [50/500], Train Loss: 0.0380, Val Loss: 0.1026, Learning Rate: 0.0080
Epoch [100/500], Train Loss: 0.0226, Val Loss: 0.0967, Learning Rate: 0.0064
Early stopping on epoch 103
FOLD 5
Epoch [50/500], Train Loss: 0.0335, Val Loss: 0.0443, Learning Rate: 0.0080
Epoch [100/500], Train Loss: 0.0173, Val Loss: 0.0310, Learning Rate: 0.0064
Epoch [150/500], Train Loss: 0.0126, Val Loss: 0.0307, Learning Rate: 0.0051
Early stopping on epoch 166

# Plotting the training and validation loss
plt.figure(figsize=(10, 6))

# This variable keeps track of the last epoch of the previous fold
last_epoch = 0
colors = plt.cm.jet(np.linspace(0, 1, k_folds))  # Generate distinct colors for each fold

for fold, (train_loss, val_loss) in enumerate(zip(all_train_losses, all_val_losses)):
    epochs = range(last_epoch, last_epoch + len(train_loss))
    plt.plot(epochs, train_loss, label=f"Training - Fold {fold + 1}", linestyle='-', color=colors[fold], alpha=0.9)
    plt.plot(epochs, val_loss, label=f"Validation - Fold {fold + 1}", linestyle='--', color=colors[fold], alpha=0.5)
    last_epoch = epochs[-1] + 1

plt.xlabel("Combined Epochs Across Folds")
plt.ylabel("Loss")
plt.title("Training and Validation Loss Over Combined Epochs")
plt.legend(loc='upper center', bbox_to_anchor=(0.5, 1.2), ncol=k_folds, fancybox=True, shadow=True)
plt.grid(True)
plt.tight_layout()
plt.show()

import numpy as np

model.load_state_dict(best_model_weights)

model.cpu()

x,y=np.meshgrid(np.linspace(-1.15,1.15,30),np.linspace(-1.15,1.15,30))
xy=np.concatenate((x.reshape(-1,1),y.reshape(-1,1)),axis=1)
z=model(torch.tensor(xy).float()).detach().numpy()
z=np.argmax(z,1).reshape(30,30)

# Define a normalization based on label values
norm = plt.Normalize(train_data['label'].min(), train_data['label'].max())

# Generate the visualization of the predictions
plt.figure(figsize=(8, 6))
plt.contourf(x, y, z, cmap="jet", alpha=0.55, norm=norm)

# Overlay scatter plot with the true labels
for label in sorted(train_data['label'].unique()):
    subset = train_data[train_data['label'] == label]
    plt.scatter(subset['feature1'], subset['feature2'], label=f'Label {int(label)}', norm=norm, 
                c=subset['label'], cmap="jet",alpha=0.7, edgecolors='w')

# Add legend and text
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")
plt.xlim(-1.15, 1.15)
plt.ylim(-1.15, 1.15)
plt.title("Model Performance on Training Data with Predicted Background")
plt.legend(loc='upper right')
plt.grid(True)

plt.show()

import numpy as np

model.load_state_dict(best_model_weights)

model.cpu()

x,y=np.meshgrid(np.linspace(-1.15,1.15,30),np.linspace(-1.15,1.15,30))
xy=np.concatenate((x.reshape(-1,1),y.reshape(-1,1)),axis=1)
z=model(torch.tensor(xy).float()).detach().numpy()
z=np.argmax(z,1).reshape(30,30)

# Define a normalization based on label values
norm = plt.Normalize(test_data['label'].min(), test_data['label'].max())

# Generate the visualization of the predictions
plt.figure(figsize=(8, 6))
plt.contourf(x, y, z, cmap="jet", alpha=0.55, norm=norm)

# Overlay scatter plot with the true labels
for label in sorted(test_data['label'].unique()):
    subset = test_data[test_data['label'] == label]
    plt.scatter(subset['feature1'], subset['feature2'], label=f'Label {int(label)}', norm=norm, 
                c=subset['label'], cmap="jet",alpha=0.7, edgecolors='w')

# Add legend and text
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")
plt.xlim(-1.15, 1.15)
plt.ylim(-1.15, 1.15)
plt.title("Model Performance on Testing Data with Predicted Background")
plt.legend(loc='upper right')
plt.grid(True)

plt.show()

Assignment 1¶

Part 1:¶

Part 2: Writing a PyTorch dataset¶

Part 3:¶